package com.leilei.count;

import com.alibaba.fastjson.JSON;
import com.leilei.Location;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.UUID;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/14 22:15
 * @desc 有 key窗口 ，使用了keyBy后你的窗口计算会由多个task执行，具有相同类型的key会进入同一个task执行。
 */
public class CountKeyWindowFunction extends RichWindowFunction<Location,String,Integer, GlobalWindow> {
    String uuid;
    /**
     *
     * @param key 数据key
     * @param window window
     * @param locations 入栈数据类型迭代器 如果是count 窗口则是count中指定的数量
     * @param out 出站数据收集器 泛型为出站数据类型
     */
    @Override
    public void apply(Integer key, GlobalWindow window, Iterable<Location> locations, Collector<String> out) {
        for (Location location : locations) {
            if (location.getGpsSpeed()>location.getLimitSpeed()) {
                out.collect(JSON.toJSONString(location));
            }
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        uuid = UUID.randomUUID().toString();
        System.out.println(uuid + "窗口打开了");
    }
}
